lst-function-definition、lst-early-return 与 lst-google-docstring-check,定义可验证自定义函数。使用 def 关键字定义函数:
def:定义函数的关键字snake_case 命名,动词开头def demonstrate_call_process(x, y):
"""
演示函数调用的完整过程
"""
print(f'接收到参数: x={x}, y={y}')
# 执行函数逻辑
result = x + y
print(f'计算结果: {result}')
# 返回结果
return result
# 函数调用过程分解
# 1. 创建栈帧(Stack Frame):为函数调用分配内存空间
# 2. 参数传递:将实参值绑定到形参名
# 3. 执行函数体:按顺序执行函数内语句
# 4. 返回值:将结果返回给调用者
# 5. 销毁栈帧:释放函数调用占用的内存
a, b = 10, 20
print('准备调用函数...')
# 函数调用:程序跳转到函数体执行
value = demonstrate_call_process(a, b)
print(f'函数返回值: {value}')
print('函数调用结束,继续执行主程序')准备调用函数...
接收到参数: x=10, y=20
计算结果: 30
函数返回值: 30
函数调用结束,继续执行主程序
pass 与 ... 只演示“先搭框架”的语法,不是练习答案。业务分析写出物必须以有输入、有输出、可验证的完整函数替换占位实现。
最基础的参数类型,按照定义时的顺序传递:
def calculate_profit(revenue, cost):
"""
计算利润的函数
参数:
revenue (float): 营业收入
cost (float): 营业成本
返回:
float: 利润额
"""
profit = revenue - cost
return profit
# 调用时必须按照定义的顺序传递参数
profit1 = calculate_profit(1000, 600)
print(f'利润1: {profit1}') # 输出: 利润1: 400
# 如果顺序错误,会得到错误结果
profit2 = calculate_profit(600, 1000)
print(f'利润2: {profit2}') # 输出: 利润2: -400 (错误!)利润1: 400
利润2: -400
通过参数名传递,不受顺序限制,代码更清晰:
def calculate_profit(revenue, cost):
"""计算利润"""
return revenue - cost
# 使用关键字参数,顺序可以任意
profit1 = calculate_profit(revenue=1000, cost=600)
profit2 = calculate_profit(cost=600, revenue=1000)
print(f'利润1: {profit1}') # 输出: 利润1: 400
print(f'利润2: {profit2}') # 输出: 利润2: 400 (正确!)
# 关键字参数提高了代码的可读性
# 尤其当函数有多个参数时,不容易混淆利润1: 400
利润2: 400
为参数提供默认值,使函数调用更灵活:
def calculate_tax(salary, rate=0.05, insurance_rate=0.02):
"""
计算税后工资
参数:
salary (float): 税前工资
rate (float): 税率,默认5%
insurance_rate (float): 社保扣除比例,默认2%
返回:
float: 税后工资
"""
# 计算税金
tax = salary * rate
# 计算社保扣除
insurance = salary * insurance_rate
# 计算实发工资
net_salary = salary - tax - insurance
return net_salary
# 情况1:只提供必填参数
net1 = calculate_tax(5000)
print(f'税后工资1(默认税率): {net1:.2f}元')
# 情况2:覆盖默认税率
net2 = calculate_tax(5000, rate=0.10)
print(f'税后工资2(10%税率): {net2:.2f}元')
# 情况3:使用关键字参数指定特定参数
net3 = calculate_tax(5000, insurance_rate=0.05)
print(f'税后工资3(5%社保): {net3:.2f}元')
# 情况4:所有参数都指定
net4 = calculate_tax(5000, rate=0.08, insurance_rate=0.03)
print(f'税后工资4(全部指定): {net4:.2f}元')税后工资1(默认税率): 4650.00元
税后工资2(10%税率): 4400.00元
税后工资3(5%社保): 4500.00元
税后工资4(全部指定): 4450.00元
使用列表等可变对象作为默认参数时要特别小心:
# 错误示例:使用列表作为默认参数
def append_item(item, items=[]): # 危险!
"""向列表添加项目"""
items.append(item)
return items
# 第一次调用
result1 = append_item('A')
print(f'第一次: {result1}') # 输出: ['A']
# 第二次调用
result2 = append_item('B')
print(f'第二次: {result2}') # 输出: ['A', 'B'] 意外!
# 问题:默认列表在函数定义时创建,所有调用共享同一个列表
# 正确做法:使用None作为默认值
def append_item_correct(item, items=None):
"""向列表添加项目(正确版本)"""
if items is None: # 每次调用都创建新列表
items = []
items.append(item)
return items
# 测试正确版本
result3 = append_item_correct('A')
print(f'正确第一次: {result3}') # 输出: ['A']
result4 = append_item_correct('B')
print(f'正确第二次: {result4}') # 输出: ['B'] 符合预期!第一次: ['A']
第二次: ['A', 'B']
正确第一次: ['A']
正确第二次: ['B']
*args接收任意数量的位置参数:
def sum_all(*args):
"""
计算所有传入数字的和
参数:
*args: 可变数量的位置参数
返回:
float/int: 所有参数的和
"""
print(f'接收到 {len(args)} 个参数: {args}')
# args是一个元组,包含所有位置参数
total = sum(args)
return total
# 调用示例
result1 = sum_all(1, 2, 3)
print(f'和1: {result1}') # 输出: 和1: 6
result2 = sum_all(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
print(f'和2: {result2}') # 输出: 和2: 55
# 也可以传递列表或元组,使用*解包
numbers = [10, 20, 30, 40]
result3 = sum_all(*numbers) # *将列表解包为独立参数
print(f'和3: {result3}') # 输出: 和3: 100接收到 3 个参数: (1, 2, 3)
和1: 6
接收到 10 个参数: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
和2: 55
接收到 4 个参数: (10, 20, 30, 40)
和3: 100
**kwargs接收任意数量的关键字参数:
def create_portfolio(**kwargs):
"""
创建投资组合
参数:
**kwargs: 股票代码和持仓数量的键值对
返回:
dict: 投资组合字典
"""
print('构建投资组合...')
# kwargs是一个字典,包含所有关键字参数
for symbol, shares in kwargs.items():
print(f' {symbol}: {shares}股')
return kwargs
# 调用示例
portfolio1 = create_portfolio(AAPL=100, MSFT=50, GOOGL=30)
print(f'组合1: {portfolio1}')
# 也可以传递字典,使用**解包
stocks = {'600519.SH': 500, '000858.SZ': 1000}
portfolio2 = create_portfolio(**stocks) # **将字典解包为关键字参数
print(f'组合2: {portfolio2}')构建投资组合...
AAPL: 100股
MSFT: 50股
GOOGL: 30股
组合1: {'AAPL': 100, 'MSFT': 50, 'GOOGL': 30}
构建投资组合...
600519.SH: 500股
000858.SZ: 1000股
组合2: {'600519.SH': 500, '000858.SZ': 1000}
多种参数类型同时使用时,必须遵循顺序:
def complex_function(a, b, c=10, *args, **kwargs):
"""
演示各种参数的组合使用
参数顺序规则:
1. 位置参数 (a, b)
2. 默认参数 (c=10)
3. 可变位置参数 (*args)
4. 可变关键字参数 (**kwargs)
"""
print(f'a={a}, b={b}, c={c}')
print(f'args: {args}')
print(f'kwargs: {kwargs}')
# 调用示例
complex_function(1, 2, 3, 4, 5, x=100, y=200)
# 输出:
# a=1, b=2, c=3
# args: (4, 5)
# kwargs: {'x': 100, 'y': 200}a=1, b=2, c=3
args: (4, 5)
kwargs: {'x': 100, 'y': 200}
def calculate_return(initial, final):
"""
计算简单收益率
参数:
initial (float): 初始价格
final (float): 最终价格
返回:
float: 收益率(小数形式)
"""
return (final - initial) / initial
# 使用示例
initial_price = 100
final_price = 115
return_rate = calculate_return(initial_price, final_price)
print(f'收益率: {return_rate:.2%}') # 输出: 收益率: 15.00%收益率: 15.00%
Python 函数可以返回多个值(实际是元组):
以下 7 个价格点代表一年内 6 个等长收益区间,因此设 periods_per_year=6;年无风险利率先按复利转换到同一期间频率,再计算年化 Sharpe。
import math, statistics # 导入数学与统计工具以统一收益风险指标的年化频率
def calculate_return_metrics(prices, periods_per_year, annual_risk_free_rate): # 定义带频率约定的收益风险函数
"""返回总收益率、路径最大回撤、年化收益率、年化波动率和年化 Sharpe。"""
if len(prices) < 3 or periods_per_year <= 0 or annual_risk_free_rate <= -1 or any(price <= 0 for price in prices): # 核验价格、区间数与无风险利率口径
raise ValueError('至少需要3个正价格,且频率为正、年无风险利率大于-100%') # 阻止无效输入生成误导性风险指标
return_intervals = len(prices) - 1 # 用相邻价格之间的区间数而非观察点数年化
period_returns = [prices[index] / prices[index - 1] - 1 for index in range(1, len(prices))] # 计算同频区间收益率
total_return = prices[-1] / prices[0] - 1 # 计算整个观察窗口的累计收益率
annualized_return = (1 + total_return) ** (periods_per_year / return_intervals) - 1 # 按实际收益区间数年化
running_peak = prices[0] # 初始化时序路径的运行峰值
max_drawdown = 0.0 # 初始化峰值到后续谷值的最大回撤
for current_price in prices: # 按时间顺序检查每个价格点
running_peak = max(running_peak, current_price) # 仅允许历史及当前价格更新运行峰值
current_drawdown = (running_peak - current_price) / running_peak # 计算当前点相对先前峰值的回撤
max_drawdown = max(max_drawdown, current_drawdown) # 保留路径上最大的峰谷损失比例
period_volatility = statistics.stdev(period_returns) # 用区间收益率而非价格水平计算波动率
annualized_volatility = period_volatility * math.sqrt(periods_per_year) # 按同一频率年化收益率波动
period_risk_free_rate = (1 + annual_risk_free_rate) ** (1 / periods_per_year) - 1 # 将年无风险利率复利转换为区间利率
sharpe_ratio = None if period_volatility == 0 else (statistics.mean(period_returns) - period_risk_free_rate) / period_volatility * math.sqrt(periods_per_year) # 用同频超额收益与波动率计算年化 Sharpe
return total_return, max_drawdown, annualized_return, annualized_volatility, sharpe_ratio # 返回五项口径一致的业务指标
prices = [100, 105, 102, 108, 112, 110, 115] # 设置一年内六个等长区间的课堂价格路径
metrics = calculate_return_metrics(prices, periods_per_year=6, annual_risk_free_rate=0.03) # 按六期频率和3%年无风险利率计算指标
assert math.isclose(metrics[1], 3 / 105, rel_tol=1e-12) and math.isclose(metrics[2], metrics[0], rel_tol=1e-12) # 验证路径回撤与六区间年化不变量
assert calculate_return_metrics([90, 100, 110], 2, 0.03)[1] == 0.0 # 验证先低后高的路径不会制造回撤
print(f'总收益率: {metrics[0]:.2%}\n最大回撤: {metrics[1]:.2%}\n年化收益率: {metrics[2]:.2%}\n年化波动率: {metrics[3]:.2%}\n年化Sharpe: {metrics[4]:.2f}') # 输出同一频率合同下的五项指标总收益率: 15.00%
最大回撤: 2.86%
年化收益率: 15.00%
年化波动率: 9.19%
年化Sharpe: 1.25
函数可以在多个地方返回,处理边界条件:
def calculate_yearly_return(prices, periods_per_year): # 定义带明确期间频率的年化收益函数
"""按相邻价格形成的收益区间数计算年化收益率,无效输入返回 None。"""
if len(prices) < 2: # 价格观察不足时无法形成收益区间
print('错误:至少需要2个价格点') # 向学习者说明缺少可比区间
return None # 在无可比期间时提前停止
if periods_per_year <= 0 or any(price <= 0 for price in prices): # 核验频率与全部价格均可用于复利计算
print('错误:频率必须为正且所有价格必须为正数') # 向学习者说明复利口径无效
return None # 在复利口径无效时提前停止
return_intervals = len(prices) - 1 # 将观察点数量转换为相邻收益区间数
yearly_return = (prices[-1] / prices[0]) ** (periods_per_year / return_intervals) - 1 # 按区间数和每年频率年化
return yearly_return # 返回频率口径明确的年化收益率
test1 = calculate_yearly_return([100, 105, 110], periods_per_year=2) # 用一年两个等长区间验证10%年收益
test2 = calculate_yearly_return([100], periods_per_year=2) # 验证观察点不足的提前返回
test3 = calculate_yearly_return([0, 100], periods_per_year=1) # 验证非正价格的提前返回
assert math.isclose(test1, 0.10, rel_tol=1e-12) and test2 is None and test3 is None # 验证区间年化和两类输入守卫
print(f'测试1(正常): {test1:.2%}\n测试2(数据不足): {test2}\n测试3(价格为0): {test3}') # 汇总展示三条执行路径错误:至少需要2个价格点
错误:频率必须为正且所有价格必须为正数
测试1(正常): 10.00%
测试2(数据不足): None
测试3(价格为0): None
没有 return 语句的函数返回 None:
def log_message(message):
"""记录日志消息"""
print(f'[LOG] {message}')
# 没有return语句,返回None
result = log_message('系统启动')
print(f'返回值: {result}') # 输出: 返回值: None
print(f'返回值类型: {type(result)}') # 输出: <class 'NoneType'>
def explicit_none():
"""显式返回None"""
print('执行完毕')
return None # 显式返回None
result2 = explicit_none()
print(f'返回值2: {result2}') # 输出: 返回值2: None[LOG] 系统启动
返回值: None
返回值类型: <class 'NoneType'>
执行完毕
返回值2: None
递归是函数调用自身的编程技巧。正确的递归必须包含:
def factorial(n):
"""
计算阶乘:n! = n × (n-1) × ... × 2 × 1
参数:
n (int): 非负整数
返回:
int: n的阶乘
递归关系:
- 0! = 1 (基准情况)
- n! = n × (n-1)! (递归情况)
"""
# 基准情况:递归的终止条件
if n <= 1:
return 1
# 递归情况:将问题分解
# n! = n × (n-1)!
return n * factorial(n - 1)
# 使用示例
print(f'0! = {factorial(0)}') # 输出: 0! = 1
print(f'5! = {factorial(5)}') # 输出: 5! = 120
print(f'10! = {factorial(10)}') # 输出: 10! = 3628800
# 递归执行过程演示(以factorial(3)为例)
# factorial(3)
# → 3 * factorial(2)
# → 2 * factorial(1)
# → 1 (基准情况)
# = 2 * 1 = 2
# = 3 * 2 = 60! = 1
5! = 120
10! = 3628800
估值时点为 \(t=0\),列表第一个元素是 \(t=1\) 年末现金流;因此每笔现金流都至少折现一期。
import math # 导入数学工具以核验平价债券的现值不变量
def present_value(cash_flows, discount_rate, period=1): # 从估值日后的第一期开始递归折现
"""
使用递归计算现金流的现值
参数:
cash_flows (list): 现金流列表
discount_rate (float): 折现率
period (int): 当前年末期数,默认从1开始(内部使用)
返回:
float: 所有现金流的现值
递归关系:
PV = CF₁/(1+r) + CF₂/(1+r)² + CF₃/(1+r)³ + ...
"""
if period > len(cash_flows): # 超过最后一个年末现金流时结束递归
return 0.0 # 返回零以终止剩余现值累加
current_cash_flow = cash_flows[period - 1] # 将第一个列表元素映射到第一年末
discount_factor = (1 + discount_rate) ** period # 按年末期数计算折现因子
current_present_value = current_cash_flow / discount_factor # 折现当前年末现金流
remaining_present_value = present_value(cash_flows, discount_rate, period + 1) # 递归折现后续年末现金流
return current_present_value + remaining_present_value # 汇总当前与后续现金流现值
cash_flows = [50, 50, 1050] # 设置票息率等于折现率的三年期平价债券现金流
rate = 0.05 # 设置与5%票息率一致的年折现率
pv = present_value(cash_flows, rate) # 从第一年末开始计算债券现值
assert math.isclose(pv, 1000.0, rel_tol=1e-12) # 验证票息率等于折现率时债券价格等于面值
print(f'现值: {pv:.2f}元') # 输出与递归讲解一致的1000元现值
# 递归调用过程:
# present_value([50, 50, 1050], 0.05, 1)
# → 50/(1.05)¹ + present_value([50, 50, 1050], 0.05, 2)
# → 50/(1.05)² + present_value([50, 50, 1050], 0.05, 3)
# → 1050/(1.05)³ + present_value([50, 50, 1050], 0.05, 4)
# → 0 (基准情况)
# = 1000.0现值: 1000.00元
import sys
# 查看默认递归深度限制
print(f'递归深度限制: {sys.getrecursionlimit()}')
# 尝试超过限制会导致错误
def deep_recursion(n):
"""深度递归演示"""
if n <= 0:
return 0
return n + deep_recursion(n - 1)
# 这会导致 RecursionError
try:
result = deep_recursion(2000)
except RecursionError as e:
print(f'错误:递归深度超限 - {e}')
# 解决方法:使用循环代替递归
def iterative_sum(n):
"""循环版本,无深度限制"""
total = 0
for i in range(n + 1):
total += i
return total
result = iterative_sum(2000)
print(f'循环版本结果: {result}')递归深度限制: 3000
循环版本结果: 2001000
Python 使用 LEGB 规则查找变量:
| 层级 | 名称 | 含义 |
|---|---|---|
| L | Local | 局部作用域(函数内部) |
| E | Enclosing | 嵌套函数的外层函数作用域 |
| G | Global | 全局作用域(模块级别) |
| B | Built-in | 内置作用域(Python 内置模块) |
# 全局变量(G层)
x = 10 # 全局变量x
y = 100 # 全局变量y
def outer_function():
"""外层函数(E层)"""
# 外层函数的局部变量
y = 20 # 遮蔽全局变量y
def inner_function():
"""内层函数(L层)"""
# 内层函数的局部变量
z = 5 # 局部变量z
# 访问各层变量
print(f'内层局部变量 z: {z}')
print(f'外层变量 y: {y}')
print(f'全局变量 x: {x}')
# 调用内层函数
inner_function()
# 调用外层函数
outer_function()
# 在全局作用域访问x和y
print(f'全局变量 x: {x}')
print(f'全局变量 y: {y}')
# 尝试访问局部变量会报错
try:
print(z)
except NameError as e:
print(f'错误:局部变量在外部不可访问 - {e}')内层局部变量 z: 5
外层变量 y: 20
全局变量 x: 10
全局变量 x: 10
全局变量 y: 100
错误:局部变量在外部不可访问 - name 'z' is not defined
在函数内部修改全局变量:
# 全局变量
counter = 0
total_sum = 0
def increment_counter():
"""增加计数器"""
global counter # 声明使用全局变量
counter += 1 # 修改全局变量
print(f'计数器: {counter}')
def add_to_total(value):
"""累加到总和"""
global total_sum # 声明使用全局变量
total_sum += value
print(f'累加后总和: {total_sum}')
# 测试
print(f'初始counter: {counter}')
increment_counter() # counter变为1
increment_counter() # counter变为2
print(f'最终counter: {counter}')
print(f'初始total_sum: {total_sum}')
add_to_total(100) # total_sum变为100
add_to_total(200) # total_sum变为300
print(f'最终total_sum: {total_sum}')初始counter: 0
计数器: 1
计数器: 2
最终counter: 2
初始total_sum: 0
累加后总和: 100
累加后总和: 300
最终total_sum: 300
嵌套函数中修改外层函数的变量:
def make_accumulator():
"""累加器工厂函数"""
# 外层函数的局部变量
total = 0
def accumulator(value):
"""内层累加函数"""
nonlocal total # 声明使用外层变量
total += value
return total
return accumulator # 返回内层函数
# 创建累加器
acc1 = make_accumulator()
acc2 = make_accumulator()
# 每个累加器有独立的状态
print(f'acc1累加10: {acc1(10)}') # 输出: 10
print(f'acc1累加20: {acc1(20)}') # 输出: 30
print(f'acc2累加100: {acc2(100)}') # 输出: 100 (独立的total)acc1累加10: 10
acc1累加20: 30
acc2累加100: 100
lambda 是创建匿名函数的简洁方式:
# lambda语法:lambda 参数列表: 表达式
# 特点:单个表达式,自动返回结果
# 示例1:简单平方函数
square = lambda x: x**2
print(f'5的平方: {square(5)}') # 输出: 5的平方: 25
# 示例2:等价于以下普通函数
def square_normal(x):
return x**2
print(f'6的平方: {square_normal(6)}') # 输出: 6的平方: 36
# 示例3:多参数lambda
add = lambda x, y: x + y
print(f'3 + 5 = {add(3, 5)}') # 输出: 3 + 5 = 8
# 示例4:lambda在金融中的应用 - 计算投资净值
net_value = lambda prices, shares: sum(p * s for p, s in zip(prices, shares))
prices = [10.5, 20.3, 15.8]
shares = [100, 200, 150]
value = net_value(prices, shares)
print(f'投资净值: {value:.2f}元')5的平方: 25
6的平方: 36
3 + 5 = 8
投资净值: 7480.00元
# 场景1:与map()结合 - 对序列中每个元素应用函数
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x**2, numbers))
print(f'平方列表: {squared}') # 输出: [1, 4, 9, 16, 25]
# 场景2:与filter()结合 - 筛选符合条件的元素
stocks = [
('AAPL', 150),
('MSFT', 280),
('GOOGL', 2700)
]
# 筛选价格大于200的股票
expensive = list(filter(lambda s: s[1] > 200, stocks))
print(f'高价股票: {expensive}') # 输出: [('MSFT', 280), ('GOOGL', 2700)]
# 场景3:与sorted()结合 - 自定义排序
portfolio = [
{'symbol': 'AAPL', 'shares': 100},
{'symbol': 'MSFT', 'shares': 50},
{'symbol': 'GOOGL', 'shares': 200}
]
# 按持股数量排序
sorted_by_shares = sorted(portfolio, key=lambda x: x['shares'])
print(f'按持股排序:')
for item in sorted_by_shares:
print(f" {item['symbol']}: {item['shares']}股")
# 场景4:与max()和min()结合 - 基于复杂条件找极值
# 找出收益率最高的股票
stock_returns = [('AAPL', 0.15), ('MSFT', 0.20), ('GOOGL', 0.12)]
best_stock = max(stock_returns, key=lambda x: x[1])
print(f'最佳股票: {best_stock[0]}, 收益率: {best_stock[1]:.2%}')平方列表: [1, 4, 9, 16, 25]
高价股票: [('MSFT', 280), ('GOOGL', 2700)]
按持股排序:
MSFT: 50股
AAPL: 100股
GOOGL: 200股
最佳股票: MSFT, 收益率: 20.00%
lambda 只适合简短逻辑,复杂情况用普通函数:
# 不好的lambda示例:过于复杂
# calculate_tax_bad = lambda salary: salary * 0.05 if salary <= 5000 else salary * 0.10
# 更好的做法:使用普通函数
def calculate_tax(salary):
"""
计算个人所得税
参数:
salary (float): 税前工资
返回:
float: 应缴税额
"""
if salary <= 5000:
return salary * 0.05
else:
return salary * 0.10
# 使用
print(f'税额(3000): {calculate_tax(3000):.2f}元')
print(f'税额(8000): {calculate_tax(8000):.2f}元')税额(3000): 150.00元
税额(8000): 800.00元
map() 对序列中每个元素应用函数:
def calculate_return(prices):
"""计算价格序列的收益率"""
returns = []
for i in range(1, len(prices)):
ret = (prices[i] - prices[i-1]) / prices[i-1]
returns.append(ret)
return returns
# 使用map处理多个股票
stocks = {
'AAPL': [150, 152, 148, 155],
'MSFT': [280, 285, 282, 290],
'GOOGL': [2700, 2720, 2680, 2750]
}
# 对每只股票应用收益率计算
returns_dict = dict(map(
lambda item: (item[0], calculate_return(item[1])),
stocks.items()
))
print('各股票收益率:')
for symbol, returns in returns_dict.items():
print(f'{symbol}: {[f"{r:.2%}" for r in returns]}')各股票收益率:
AAPL: ['1.33%', '-2.63%', '4.73%']
MSFT: ['1.79%', '-1.05%', '2.84%']
GOOGL: ['0.74%', '-1.47%', '2.61%']
filter() 筛选符合条件的元素:
stocks = [
{'symbol': 'AAPL', 'price': 150, 'pe': 25},
{'symbol': 'MSFT', 'price': 280, 'pe': 30},
{'symbol': 'GOOGL', 'price': 2700, 'pe': 20},
{'symbol': 'T业务规定', 'price': 800, 'pe': 100}
]
# 筛选PE < 30的股票
value_stocks = list(filter(lambda s: s['pe'] < 30, stocks))
print('价值股票(PE < 30):')
for stock in value_stocks:
print(f" {stock['symbol']}: PE={stock['pe']}")价值股票(PE < 30):
AAPL: PE=25
GOOGL: PE=20
reduce() 对序列做累积计算:
from functools import reduce
def calculate_compound_return(returns):
"""
计算复合收益率
参数:
returns (list): 期收益率列表
返回:
float: 复合收益率
"""
# 复合收益率 = (1+r₁) × (1+r₂) × ... × (1+rₙ) - 1
# 使用reduce累积计算
compound = reduce(
lambda acc, r: acc * (1 + r), # 累积乘法
returns,
1.0 # 初始值
)
return compound - 1
# 示例:4个季度的收益率
quarterly_returns = [0.05, 0.03, 0.08, 0.02]
annual_return = calculate_compound_return(quarterly_returns)
print(f'年复合收益率: {annual_return:.2%}')年复合收益率: 19.14%
def calculate_sharpe_ratio(returns, periods_per_year=12, annual_risk_free_rate=0.03): # 定义带期间频率与年无风险利率合同的夏普函数
"""
计算同频收益率的年化夏普比率。
参数:
returns (list[float]): 同频小数收益率,至少两个有限值。
periods_per_year (int): 每年收益区间数。
annual_risk_free_rate (float): 年有效无风险利率。
返回:
float | None: 年化比率;零波动时返回None。
异常:
ValueError: 样本、频率或利率合同不满足时。
"""
import math # 导入数学工具以按期间频率年化夏普比率
import statistics # 导入统计工具以计算同频收益率波动
if len(returns) < 2 or periods_per_year <= 0 or annual_risk_free_rate <= -1: # 核验样本、频率与无风险利率合同
raise ValueError('至少需要2个收益率,且频率为正、年无风险利率大于-100%') # 阻止不可比口径进入计算
period_volatility = statistics.stdev(returns) # 计算同频区间收益率的样本波动率
if period_volatility == 0: # 同频收益没有波动时无法定义夏普比率
return None # 避免以零波动率作为分母
period_risk_free_rate = (1 + annual_risk_free_rate) ** (1 / periods_per_year) - 1 # 将年无风险利率复利转换到同频期间
sharpe = (statistics.mean(returns) - period_risk_free_rate) / period_volatility * math.sqrt(periods_per_year) # 年化同频超额收益风险比
return sharpe # 返回频率合同一致的年化夏普比率test_returns = [0.05, 0.03, 0.08, -0.02, 0.06] # 设置五个月度课堂收益率
sharpe_ratio = calculate_sharpe_ratio(test_returns, periods_per_year=12, annual_risk_free_rate=0.03) # 按月度频率与3%年无风险利率计算
assert math.isclose(sharpe_ratio, 3.414509694460191, rel_tol=1e-12) # 锁定同频转换后的年化夏普不变量
print(f'年化夏普比率: {sharpe_ratio:.2f}') # 输出月度收益合同下的年化结果年化夏普比率: 3.41
calculate_sharpe_ratio()| 知识点 | 核心要点 |
|---|---|
| 函数定义 | def + 函数名 + 参数 + return |
| 参数类型 | 位置、关键字、默认、*args、**kwargs |
| 返回值 | 单返回值、多返回值(元组)、None |
| 递归 | 基准情况 + 递归情况,注意深度限制 |
| 作用域 | LEGB 规则,global/nonlocal |
| lambda | 匿名函数,适合简短的一次性逻辑 |
data、Shibor、Libor、Hibor 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
import pandas as pd # 导入Pandas数据分析库
# 从Excel文件读取数据存入data
data = pd.read_excel(r"https://huoran.oss-cn-shenzhen.aliyuncs.com/20220811/xlsx/1557658167503511552.xlsx")
data.columns =["日期","Shibor","Libor","Hibor"] # 定义列表data.columns
Shibor = [] # 定义列表Shibor
for i in data["Shibor"]: # 遍历data["Shibor"]中的每个i
Shibor.append(i) # 将Shibor利率数据添加到列表
Libor = [] # 定义列表Libor
for i in data["Libor"]: # 遍历data["Libor"]中的每个i
Libor.append(i) # 将美元Libor利率数据添加到列表
Hibor = [] # 定义列表Hibor
for i in data["Hibor"]: # 遍历data["Hibor"]中的每个i
Hibor.append(i) # 将港元Hibor利率数据添加到列表
f_mean = lambda x: sum(x)/len(x) #用lambda函数定义并且参数x以列表的数据结构输入
Shibor_mean = f_mean(x=Shibor) #计算Shibor的平均值
Libor_mean = f_mean(x=Libor) #计算美元Libor的平均值
Hibor_mean = f_mean(x=Hibor) #计算美元Hibor的平均值
print("3月11日到29日期间3个月期Shibor平均值",round(Shibor_mean,6)) # 输出3月11日到29日期间3个月期Shibor平均值
print("3月11日到29日期间3个月期美元Libor平均值",round(Libor_mean,6)) # 输出3月11日到29日期间3个月期美元Libor平均值
print("3月11日到29日期间3个月期Hibor平均值",round(Hibor_mean,6)) # 输出3月11日到29日期间3个月期Hibor平均值运行后核对:核对 data、Shibor、Libor、Hibor 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。
拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。
data、Shibor、Libor、Hibor 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
import pandas as pd # 导入Pandas数据分析库
# 从Excel文件读取数据存入data
data = pd.read_excel(r"https://huoran.oss-cn-shenzhen.aliyuncs.com/20220811/xlsx/1557658167503511552.xlsx")
data.columns =["日期","Shibor","Libor","Hibor"] # 定义列表data.columns
Shibor = [] # 定义列表Shibor
for i in data["Shibor"]: # 遍历data["Shibor"]中的每个i
Shibor.append(i) # 将Shibor利率数据添加到列表
Libor = [] # 定义列表Libor
for i in data["Libor"]: # 遍历data["Libor"]中的每个i
Libor.append(i) # 将美元Libor利率数据添加到列表
Hibor = [] # 定义列表Hibor
for i in data["Hibor"]: # 遍历data["Hibor"]中的每个i
Hibor.append(i) # 将港元Hibor利率数据添加到列表
def f_sigma(x): # 定义函数f_sigma
'''通过Python定义一个计算变量波动率的函数
x:代表变量的样本值,可以用列表的数据结构输入'''
n = len(x) # 获取数据长度
u_mean = sum(x)/n #计算变量样本值的均值
z = [] #生成一个空列表
for t in range(n): # 遍历range(n)中的每个t
z.append((x[t]-u_mean)**2) # 将计算结果累加到z列表
return (sum(z)/(n-1))**0.5 #计算波动率
Shibor_sigma = f_sigma(x=Shibor) #计算Shibor波动率
Libor_sigma = f_sigma(x=Libor) #计算Libor波动率
Hibor_sigma = f_sigma(x=Hibor) #计算Hibor波动率
print("3月11日到29日期间3个月期Shibor波动率",round(Shibor_sigma,6)) # 输出3月11日到29日期间3个月期Shibor波动率
print("3月11日到29日期间3个月期没有Libor波动率",round(Libor_sigma,6)) # 输出3月11日到29日期间3个月期没有Libor波动率
print("3月11日到29日期间3个月期Hibor波动率",round(Hibor_sigma,6)) # 输出3月11日到29日期间3个月期Hibor波动率运行后核对:核对 data、Shibor、Libor、Hibor 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。
拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。
Shibor_mean、Libor_mean、Hibor_mean、Shibor_sigma 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。Hibor_sigma 的结果;同时写出方向、数量级或表格/图形结构。# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
Shibor_mean = 0.028065 # 设置期数/数量为0
Libor_mean = 0.026077 # 设置期数/数量为0
Hibor_mean = 0.017619 # 设置期数/数量为0
Shibor_sigma = 0.000286 # 设置数组长度参数为0
Libor_sigma = 0.000111 # 设置数组长度参数为0
Hibor_sigma = 0.000702 # 设置数组长度参数为0
print(Shibor_mean > Libor_mean) #判断Shibor的平均值是否大于美元Libor的平均值
print(Libor_mean > Hibor_mean) #判断美元Libor的平均值是否大于Hibor的平均值
print(Shibor_sigma > Libor_sigma) #判断Shibor的波动率是否大于Libor的波动率
print(Libor_sigma > Hibor_sigma) #判断美云Libor的波动率是否大于Hibor的波动率
print(Shibor_sigma > Hibor_sigma) #判断Shibor的波动率是否大于Hibor的波动率运行后核对:核对 Shibor_mean、Libor_mean、Hibor_mean、Shibor_sigma 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。
拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。
lst-function-definition、lst-early-return 与 lst-google-docstring-check,定义可验证自定义函数。cash_flows 与折现率 discount_rate;折现率必须大于 −1,现金流不得为空。def calculate_dcf(cash_flows,discount_rate): # 定义现金流折现合同
"""返回从t=0开始的现金流现值;拒绝空输入和不合法利率。""" # 说明时间口径与失败分支
if not cash_flows: raise ValueError('cash_flows不能为空') # 空序列无法形成估值依据
if discount_rate<=-1: raise ValueError('discount_rate必须大于-1') # 避免折现因子为零或符号失真
return sum(cash_flow/(1+discount_rate)**period for period,cash_flow in enumerate(cash_flows)) # 按t=0口径逐期折现
two_period_value=calculate_dcf([-100,110],.10) # 构造可手算的两期现金流
assert abs(two_period_value-0)<1e-10 # 用-100+110/1.1=0另行核对
print(round(two_period_value,2)) # 输出两位小数参考结果t=0 时间口径;空输入和 r≤−1 通过同一接口请先检查数据后再继续。0.00;空列表和利率为 −1 的情况应分别报错。t=0/t=1、遗漏空输入、允许 r≤−1。[商业大数据分析与应用]